package com.linshidream.intrachat.simplenio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Created on 2025/7/30 18:54
 *
 * @author linshidream
 * @version 1.0.0
 * @description
 */

public class EventLoopServer {

    public static void main(String[] args) throws IOException {
        WorkerServer bossWorkerServer = new WorkerServer();
        // 绑定端口并开启4个 Worker 线程
        bossWorkerServer.bind(10010).start(4);
    }

    public static class WorkerServer {

        private int port;

        private int index = 0;

        public WorkerServer bind(int port) {
            this.port = port;
            return this;
        }

        /**
         * 启动监听服务
         *
         * @param workerCount
         * @throws IOException
         */
        public void start(int workerCount) throws IOException {
            // 初始化 Worker 数量
            Worker[] workers = new Worker[workerCount];
            for (int i = 0; i < workerCount; i++) {
                workers[i] = new Worker();
                String name = "server-worker-" + i;
                new Thread(workers[i], name).start();
                System.out.println("服务端开启了 Worker[" + i + "]线程：" + name);
            }

            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.bind(new InetSocketAddress(port));

            // bossSelector 主线程只负责监听连接
            Selector bossSelector = Selector.open();
            serverSocketChannel.register(bossSelector, SelectionKey.OP_ACCEPT);

            System.out.println("EventLoop 服务端启动成功，监听端口: " + port + ", Worker 线程数量：" + workerCount);

            while (true) {
                bossSelector.select();
                Iterator<SelectionKey> iterator = bossSelector.selectedKeys().iterator();

                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    iterator.remove();

                    if (selectionKey.isAcceptable()) {
                        ServerSocketChannel sc = (ServerSocketChannel) selectionKey.channel();
                        SocketChannel client = sc.accept();
                        System.out.println("客户端已连接成功:" + client.getRemoteAddress());

                        // 轮询策略选择 Worker 线程，专门负责监听读写
                        Worker worker = workers[index++ % workerCount];
                        worker.register(client);
                    }
                }
            }
        }
    }

    /**
     * Worker 工作线程
     */
    public static class Worker implements Runnable {

        // 每个 Worker 对象持有一个监听器
        private final Selector selector;

        public Worker() throws IOException {
            this.selector = Selector.open();
        }

        // 存放 Socket 通道代码
        private final BlockingQueue<SocketChannel> socketChannels = new LinkedBlockingDeque<>();

        public void register(SocketChannel sc) {
            socketChannels.offer(sc);
            // 这里一定要唤醒，run 方法里面就开始取 SocketChannel 去注册
            selector.wakeup();
        }

        /**
         * 读取客户端消息并响应 Echo
         *
         * @param channel
         * @throws IOException
         */
        public static void readAndRespond(SocketChannel channel) throws IOException {
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            int len = channel.read(buffer);
            if (len == -1) {
                return;
            }
            buffer.flip();
            String reqMsg = new String(buffer.array(), 0, len, StandardCharsets.UTF_8);
            System.out.println("服务端收到客户端消息:" + reqMsg);

            String respMsg = "echo:" + reqMsg;
            ByteBuffer req = ByteBuffer.wrap((respMsg + "\n").getBytes(StandardCharsets.UTF_8));
            channel.write(req);

            System.out.println("服务端响应客户端消息:" + respMsg);
        }

        @Override
        public void run() {
            try {
                while (true) {
                    selector.select();

                    // 会把所有的 SocketChannel 注册到 selector，直到注册完成
                    SocketChannel sc;
                    while ((sc = socketChannels.poll()) != null) {
                        sc.configureBlocking(false);
                        sc.register(selector, SelectionKey.OP_READ);
                    }

                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();

                    while (iterator.hasNext()) {
                        SelectionKey selectionKey = iterator.next();
                        iterator.remove();

                        if (selectionKey.isReadable()) {
                            SocketChannel client = (SocketChannel) selectionKey.channel();
                            readAndRespond(client);
                        }
                    }
                }
            } catch (Exception e) {
                System.out.println("Worker 线程执行异常," + Thread.currentThread().getName());
            }
        }
    }
}
